package com.orion.config;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.listener.BatchErrorHandler;

import java.util.HashMap;
import java.util.Map;

//@Component
public class KafkaBatchExceptionHandler implements BatchErrorHandler {
    @Override
    public void handle(Exception e, ConsumerRecords<?, ?> consumerRecords) {
        System.out.println("do nothing");
    }

    @Override
    public void handle(Exception thrownException, ConsumerRecords<?, ?> consumerRecords, Consumer<?, ?> consumer) {
        Map<TopicPartition, Long> offsetsToReset = new HashMap<>(16);
        consumerRecords.forEach((item) -> {
            System.out.println("batch error handler will seek min offset , data : " + item + " , exception : " + thrownException);
            offsetsToReset.compute(new TopicPartition(item.topic(), item.partition()),
                    (k, v) -> v == null ? item.offset() : Math.min(v, item.offset()));
        });
        offsetsToReset.forEach((k, v) -> consumer.seek(k, v));
    }
}